package org.jetlinks.pro.messaging.rabbitmq.impl;

import com.rabbitmq.client.ConnectionFactory;
import org.jetlinks.pro.datasource.CommandHandler;
import org.jetlinks.pro.datasource.DataSource;
import org.jetlinks.pro.messaging.rabbitmq.RabbitUtils;
import org.jetlinks.pro.messaging.rabbitmq.command.CreateNativeSender;
import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.RabbitFlux;
import reactor.rabbitmq.Sender;
import reactor.rabbitmq.SenderOptions;

public class CreateNativeSenderHandler implements CommandHandler<CreateNativeSender, Sender> {

    @Override
    public Sender handle(DataSource dataSource, CreateNativeSender command) {

        ConnectionFactory connectionFactory = RabbitUtils.createConnectionFactory(command.getProperties());
        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(1000);
        connectionFactory.useNio();

        SenderOptions senderOptions = new SenderOptions()
            .connectionFactory(connectionFactory)
            .resourceManagementScheduler(Schedulers.boundedElastic());

        return RabbitFlux.createSender(senderOptions);
    }

}
